
#include "common.h"
#include "http.h"
#include "sock.h"
#include "cache_index.h"
#include "cache_control.h"

static const char *HttpRequestLineFormat = "%s %s HTTP/%d.%d\r\n";
static const char *HttpReplyLineFormat = "HTTP/%d.%d %d %s\r\n";


static void http_miss_handle(request_t *r);
static void http_hit_handle(request_t *r, int revalidate);
static void http_revalidate_handle(request_t *r);
/* connect handlers */
static void connect_start(int fd, struct event *ev, PF *handler, request_t *arg);
static void connect_finish(int fd, short res, void *arg);
/* request handle */
static int request_hit_state(request_t *request) ;
static int request_parse(request_t *r, char *buf, int size);
static void http_request_handle(request_t *request);
static void request_free(request_t *req);
static request_t *request_init(conn_t *con);
static void http_request_read(int fd, short flag, void *arg);
static void http_request_send(request_t *r);
static void http_request_send_finish(int fd, int error, int res, void *arg);
static int conn_server_fd_create(request_t *r);
static string *request_string_build(const request_t *r) ;
/* reply handle */
static void http_ims_reply_read(int fd, short flag, void *arg);
static void http_reply_read(int fd, short flag, void *arg);
static void reply_free(reply_t *reply);
static reply_t * reply_init(void);
static int reply_parse(reply_t *reply, const char *buf, const size_t size);
static string *reply_header_build(header_t *head, int statu, http_version_t version, int conn_state);
static void http_reply_header_send(int fd, request_t *r, int data_state);
static void http_reply_header_send_done(int fd, int error, int res, void *arg);
static void http_reply_body_send(int fd, request_t *r, int data_send_state);
static void http_reply_body_send_done(int fd, int error, int res, void *arg);
static void http_reply_send(int fd, short flag, void *data);
static int http_reply_swapout(reply_t *rep, string *file);
static void http_hit_waiting(int fd, short flag, void *data);
static int required_data_is_exist(reply_t *reply);
static void reply_in_mem_maintain(reply_t *rep) ;

/********************************************/
/*****************conn handlers *************/
void conn_timeout_set(struct event *ev, PF *handler, void *arg, int timeout, struct event_base *base)
{
	if(!ev) 
		return;
	
	struct timeval t;
	t.tv_sec = timeout;
	t.tv_usec = 0;
	/* init timeout event */
	timeout_set(ev, handler, arg);
	event_base_set(base, ev);
	if( -1 == timeout_add(ev, &t)) {
		error_log("fd %p timeout add failed\n", ev);
	}
}

conn_t *conn_new(int cfd, short cfl, PF *ch, void *arg, struct in_addr *client_addr, \
				struct event_base *base, thread_t *thread) 
{
	assert(base);

	if(cfd < 0)
		return NULL;

	conn_t *con = mempool_alloc(sizeof(*con));
	if(!con) {
		debug_log(3, "fd %d alloc failed\n", cfd);
		return NULL;
	}

	con->cfd = cfd;

	/* init all event  */
	event_set(&con->client_event, -1, -1, NULL, NULL);
	event_set(&con->server_event, -1,  -1, NULL, NULL);
	event_set(&con->timeout_event, -1, -1, NULL, NULL);
	evtimer_set(&con->timer, NULL, NULL);
	event_base_set(base, &con->timer);
	event_base_set(base, &con->client_event);
	event_base_set(base, &con->server_event);
	event_base_set(base, &con->timeout_event);
	
	/* init client event */
	if(arg) 
		event_set(&con->client_event, cfd, cfl, ch, arg);
	else 
		event_set(&con->client_event, cfd, cfl, ch, con);
	if( -1 == event_add(&con->client_event, 0)) {
		error_log("fd %d event add failed\n", cfd);
		mempool_free(con);
		return NULL;
	}
	
	con->sfd = -1;
	con->nrequests = 0;
	INIT_LIST_HEAD(&con->requests);
	INIT_LIST_HEAD(&con->list);
	if(client_addr)
		con->client_addr = *client_addr;
	else 
		memset(&con->client_addr, 0, sizeof(con->client_addr));
	memset(&con->server_addr, 0, sizeof(con->server_addr));
	con->in = string_limit_init(MAX_HEADER + MAX_URL);
	con->thread = thread;
	/* insert into thread active list */
	if(thread)
		queue_push(thread->active_conns, &con->list);
	return con;
}

void conn_free(conn_t *con)
{
	if(!con) 
		return;
	
	struct list_head *p, *n;
	
	debug_log(3, "free %p\n", con);
	event_del(&con->client_event);
	event_del(&con->server_event);
	event_del(&con->timeout_event);
	event_del(&con->timer);
	string_clean(con->in);
	sock_close(con->cfd);
	sock_close(con->sfd);
	list_for_each_safe(p, n, &con->requests) {
		request_t *req = list_entry(p, request_t, list);
		if(req)
			request_free(req);
	}
	mempool_free(con);
}

/*****************************************************/

/* first read request data */
void conn_request_read(int fd, short flag, void *arg)
{
	if(fd < 0) 
		return ;
	if(arg == NULL) {
		debug_log(3, "arg is %p, return.\n", arg);
		return ;
	}

	conn_t *con = arg;
	string *in;
	int sz;
	int ret;
	int left;
	request_t *req;
	
	if(con->nrequests == 0) 
		req = request_init(con);
	else 
		req = list_entry(con->requests.next, request_t, list);
	assert(req);
	in = con->in;
	left = in->size - in->offset;
	if(left <= 0) {
		string_reset(in);
		left = in->size  - in->offset;
	}
	sz = read(fd, in->buf + in->offset, left);
   	if(sz == -1) {
		if(error_ignore(errno)) {
			return;
		}
		conn_free(con);
		return;
	}	
	else if(sz == 0 && in->offset == 0) {
		/* no data in */
		conn_free(con);
		return;
	}
	else if(sz == 0) {
		if(req->body == NULL) 
			debug_log(5, "request in %d, process request now\n", in->offset);
		else if(req->body != NULL && req->body->offset >= req->content_length)
			debug_log(5, "request body in %d, process request now\n", req->body->offset);
		else if(req->body != NULL && req->body->offset < req->content_length)
			return;
	}

	/* update in->offset */
	in->offset += sz;
	debug_log(3, "in %d bytes, left room size %d\n", sz, in->size - in->offset);
	debug_log(3, "\n================\n""%s""\n================\n", in->buf);

	/* set timeout to client life */
	timeout_update(&con->timeout_event, conn_timeout_handle, con, conf->timeout.client_life);
	if (-1 == (ret = request_parse(req, in->buf, in->offset))) {
		debug_log(1, "fd %d incorrected request, parsed failed\n", fd);
		conn_free(con);
		return ;
	}
	cache_control_set(&req->cc, req->headers);
	/* cacheable ? */
	req->cacheable = cache_able(req);
	/* check content-length */
	if(req->body && req->content_length > 0 && req->body->offset < req->content_length) {
		/* make http_request_read to handle */
		event_update(&con->client_event, con->cfd, EV_READ | EV_PERSIST, http_request_read, req);
		debug_log(5, "register http_request_handle %p continue reading\n", http_request_read);
		return;
	}
	http_request_handle(req);
}


/**********************************************/
/* request handlers                           */

static void http_request_handle(request_t *request)
{
	assert(request);
	int state;

	if(cache_lock())
		return;
	state = request_hit_state(request);
	debug_log(3, "http://%s%s %s\n", \
					request->host, request->url_path, hit_state_string(state));
	switch(state) {
		case TCP_REVALIDATE:
		case TCP_HIT:
			http_hit_handle(request, state);
			break;
		case TCP_MISS:
			http_miss_handle(request);
			break;
		default: 
			http_miss_handle(request);
			break;
	}
	cache_unlock();
}

static int request_hit_state(request_t *r) 
{
	if(!r) 
		return TCP_MISS;

	cache_file_t *cf;
	char buf[MAX_URL];
	int res;
		
	if(r->ims) {
		r->hit_state = TCP_REVALIDATE;
		r->log_state = r->hit_state;
		return TCP_REVALIDATE;
	}
	res = snprintf(buf, MAX_URL -1, "http://%s%s", r->host, r->url_path);
	cf = cache_file_search(buf, res);
	if(NULL == cf) {
		r->hit_state = TCP_MISS;
		r->log_state = r->hit_state;
		return TCP_MISS;
	}
	if(cf->should_free) {
		r->hit_state = TCP_MISS	;
		r->log_state = r->hit_state;
		return r->hit_state;
	}
	r->cache = cache_ref(cf);
	if(cf->statu != CACHE_IN_DISK) {
		r->hit_state = TCP_HIT;
		debug_log(3, "%s downloading\n", buf);
		return TCP_HIT;
	}
	if(freshness_check(r->cache, r) == CACHE_IN_STALE) {
		r->hit_state = TCP_REVALIDATE;
		r->log_state = TCP_HIT;
		return TCP_REVALIDATE;
	}
					
	r->hit_state = TCP_HIT;
	r->log_state = TCP_HIT;
	return TCP_HIT;
}

static request_t *request_init(conn_t *con)
{
	request_t *req = mempool_alloc(sizeof(*req));
	if(!req) {
		debug_log(1, "mempool_alloc request %p failed\n", req);
		return NULL;
	}

	req->con = con;
	INIT_LIST_HEAD(&req->list);
	list_add_tail(&req->list, &con->requests);
	con->nrequests++;

	debug_log(5, "connection number of request %d\n", con->nrequests);
	return req;
}

static void request_free(request_t *r)
{
	if(!r)
		return;

	header_free(r->headers);
	safe_free(r->url_path);
	safe_free(r->host);
	reply_free(r->reply);
	cache_unref(r->cache);
	list_del(&r->list);

	mempool_free(r);
}

static int request_parse(request_t *req, char *buf, int size)
{
	assert(req);
	assert(buf);
	assert(size >= 0);

	char *start = (char *)buf;
	char *end;
	int minor;
	int major;
	size_t request_line_size = 0;
	int http_header_size = 0;  /* include request line */
	int header_size;
	char line[MAX_URL];
	int port = 80;
	char *meth;
	char host[MAX_BUFFER];
	char urlpath[MAX_BUFFER];
	char proto[MAX_BUFFER];
	header_entry_t *e;
		
	memset(host, 0, MAX_BUFFER);
	memset(urlpath, 0, MAX_BUFFER);
	memset(proto, 0, MAX_BUFFER);
	
	/* parse request line */
	if((end = (char *)memchr((char *)buf, '\n', size)) == NULL) {
		debug_log(1, "Incomplete request, waiting for end of request line.\n------\n%s\n------\n", \
					 buf);
		return -1;
	}
	
	request_line_size = end - start;
	end = (char *)memchr(start, ' ', request_line_size);
	assert(end);
	http_header_size = headers_end(buf, size);
	if(0 == http_header_size) {
		debug_log(1, "{%s} failed to find headers end\n", buf);
		return -1;
	}
	header_size = http_header_size - request_line_size;
	/* parse method */
	meth = (char *) strndup(start, end - start);
	req->method.name = meth;
	req->method.id = http_method_id_get(meth);
	if(req->method.id == METHOD_NONE) {
		debug_log(1, "method '%s' unknown\n", meth);
		safe_free(meth);
		safe_free(req);
		return -1;
	}
	debug_log(3, "method {%s, %d}\n", req->method.name, req->method.id);
	
	/* parse url path */
	while(*end == ' ' || *end == '\t')
		end++;
	start = end;
	end = strchr(start, ' ');
	assert(end);
	snprintf(line, end - start + 1, "%s", start);
	if(line[0] != '/') {
		if(sscanf(line, "%[^:]://%[^:]:%d%[^\r\n]", proto, host, &port, urlpath) < 4) {
			if(sscanf(line, "%[^:]://%[^/]%[^\r\n]", proto, host, urlpath) < 3) {
				debug_log(3, "unknown {%s}\n", line);
				return -1;
			}
		}
	
		req->proto = proto_string_to_type(proto);
		req->remote_port = port;
		req->url_path = strdup(urlpath);
		if(host[0] != '\0') 
			req->host = strdup(host);
	}
	else 
		req->url_path = strdup(line);

	debug_log(3, "urlpath %s, port %d, host %s, protocol %s\n", \
					req->url_path, req->remote_port, req->host, proto_type_to_string(req->proto));

	/* parse http version */
	start = end + 1;
	if(strncasecmp(start, "http/", 5) != 0) {
		debug_log(1, "start %s can not find HTTP/\n", start);
		safe_free(meth);
		safe_free(req->url_path);
		safe_free(req);
		return -1;
	}
	major = atoi(start + 5);
	end = strchr(start + 5, '.');
	minor = atoi(end + 1);
	req->version.major = major;
	req->version.minor = minor;
	debug_log(3, "http version %d.%d\n", major, minor);
		
	/* parse http header */
	req->headers = http_header_parse(buf + request_line_size + 1, size - request_line_size - 1);
	/* get host */
	if(!req->host) {
		if((e = header_entry_get(req->headers, HDR_HOST)))
			req->host = strdup(e->value);
	}
	/* parse Content-Length */
	if((e = header_entry_get(req->headers, HDR_CONTENT_LENGTH))) {
		req->content_length = atoi(e->value);
		if(req->content_length > 0) {
			req->body = string_limit_init(req->content_length);
			assert(req->body);
			int content = size - http_header_size;
			if(content > 0)
				string_append(req->body, buf + http_header_size, content);
			debug_log(5, "find %d bytes body\n", content);
		}
	}
	if(req->remote_port <= 0) 
		req->remote_port = 80;
	if(req->remote_port < 1 || req->remote_port > 65535) {
		debug_log(4, "invalid port %d\n", req->remote_port);
		return -1;
	}
	
	return 0;
}
/********************** end ******************/


/**********************************************/
/* reply handlers                             */
static void reply_free(reply_t *reply)
{
	if(!reply) 
		return;
	
	mem_free(reply->in);
	header_free(reply->headers);
	sock_close(reply->disk);
	disk_map_reset(&reply->map);
	mempool_free(reply);
}

/**********************************************/

/* event handle */
/* return -1 failed, 0 success. */
int event_update(struct event *ev, int fd, short new_flag, PF *handler, void *arg)
{
	assert(ev != NULL);

	struct event_base *base = ev->ev_base;
	if(event_del(ev) == -1) {
		error_log("fd %d event del failed\n", fd);
	}
	event_set(ev, fd, new_flag, handler, arg);
	event_base_set(base, ev);
	if (event_add(ev, 0) == -1) {
		error_log("fd %d event add failed\n", fd);
		return -1;
	}
	
	return 0;
}

int timeout_update(struct event *ev, PF *handler, void *arg, int timeout)
{
	assert(ev != NULL);

	struct timeval tv;
	struct event_base *base = ev->ev_base;
	tv.tv_sec = timeout;
	tv.tv_usec = 0;
	timeout_del(ev);
	timeout_set(ev, handler, arg);
	event_base_set(base, ev);
	if (timeout_add(ev, &tv) == -1) return -1;

	debug_log(3, "%d.0 timeout update\n", timeout);
	return 0;
}

int timer_update(struct event *ev, PF *handler, void *arg, int sec, int usec)
{
	assert(ev != NULL);

	struct timeval tv;
	struct event_base *base = ev->ev_base;
	tv.tv_sec = sec;
	tv.tv_usec = usec;
	evtimer_del(ev);
	evtimer_set(ev, handler, arg);
	event_base_set(base, ev);

	if (evtimer_add(ev, &tv) == -1) return -1;
	
	return 0;
}

char *hit_state_string(int state)
{
	switch(state) {
		case TCP_NONE:
			return "TCP_NONE";
		case TCP_HIT:
			return "TCP_HIT";
		case TCP_MISS:
			return "TCP_MISS";
		case TCP_REVALIDATE:
			return "TCP_REVALIDATE";
		case TCP_REFRESH_MISS:
			return "TCP_REFRESH_MISS";
		case TCP_REFRESH_HIT:
			return "TCP_REFRESH_HIT";
		case TCP_REFRESH_ERROR:
			return "TCP_REFRESH_ERROR";
		case TCP_SHM_HIT:
			return "TCP_SHM_HIT";
		case TCP_MEM_HIT:
			return "TCP_MEM_HIT";
		default: 
			return "TCP_UNKOWN";
			break;
	}
}

static void http_miss_handle(request_t *r)
{
	assert(r);
	
	conn_t *con = r->con;
	char buf[MAX_BUFFER];

	con->sfd = conn_server_fd_create(r);
	if(con->sfd == -1) 
		return;
	
	snprintf(buf, MAX_BUFFER -1 , "http://%s%s", r->host, r->url_path);
	/* create cache file. and insert into cache table */
	/* if miss, not need lock cache */
	cache_file_t *cf = cache_file_create(buf);
	r->cache = cache_ref(cf);
	r->cache->statu = CACHE_DOWNLOADING;
	if(r->cacheable == CACHE_ABLE_YES)
		cache_file_insert(cf);
	else 
		debug_log(5, "%s%s can't be cached\n", r->host, r->url_path );
	
	event_del(&con->client_event);
	/* set server event */
	event_update(&con->server_event, con->sfd, EV_WRITE | EV_PERSIST, connect_finish, r);
	timeout_update(&con->timeout_event, conn_timeout_handle, con, conf->timeout.connect);
	connect_start(con->sfd, &con->server_event, \
					connect_finish, r);
}

static void http_hit_handle(request_t *request, int state)
{
	assert(request);

	conn_t *con = request->con;
	cache_file_t *cache = request->cache;
	reply_t *rep = request->reply;
		
	if(state == TCP_REVALIDATE) 
	{
		http_revalidate_handle(request);
		return;
	}

	/* TCP_HIT */
	request->reply = reply_init();
	rep = request->reply;
	rep->header_size = cache->header_size;
	rep->body_size = cache->body_size;
	if(rep->header_size == 0 && rep->body_size == 0) {
		timer_update(&con->timer, http_hit_waiting, request, 0, 1000);
		debug_log(3, "timer %d.%d callback\n", 0, 300);
		return;
	}
	sock_cork_set(con->cfd);
	rep->swapout.offset = rep->header_size + rep->body_size;	
	if(event_update(&con->client_event, con->cfd, EV_WRITE | EV_PERSIST, http_reply_send, request) == -1) {
		debug_log(3, "fd %d event update failed\n", con->cfd);
		return ;
	}
}

static void http_revalidate_handle(request_t *request)
{
	assert(request);

	conn_t *con = request->con;

	con->sfd = conn_server_fd_create(request);
	if(con->sfd == -1) {
		conn_free(con);
		return;
	}

	/* set server event */
	event_set(&con->server_event, con->sfd, EV_WRITE | EV_PERSIST, connect_finish, request);
	event_add(&con->server_event, 0);
	timeout_update(&con->timeout_event, conn_timeout_handle, con, conf->timeout.connect);

	connect_start(con->sfd, &con->server_event, \
					connect_finish, request);
}


static void connect_start(int fd, struct event *ev, PF *handler, request_t *arg)
{
	if(fd < 0) 
		return;
	
	assert(ev);

	struct sockaddr_in srv;
	int res;
	request_t *request = arg;
	conn_t *con = request->con;

	debug_log(5, "fd %d connecting...\n", fd);
	srv.sin_family = AF_INET;
	srv.sin_port = htons(request->remote_port);
	srv.sin_addr = con->server_addr;
	if(0 == (res = connect(fd, (struct sockaddr *)&srv, sizeof(srv)))) {
		if(handler) {
			handler(fd, EV_WRITE, arg);
			return;
		}
	}
	else if(res < 0) {
		if(!error_ignore(errno)) {
			error_log("fd %d connecting failed\n", fd);
			conn_free(con);
		}
	}
}

static void connect_finish(int fd, short flag, void *arg)
{
	int err;
	int errlen = sizeof(err);
	request_t *request = arg;

	if(0 != getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &errlen)) {
		error_log("fd %d connect failed\n", fd);
		conn_free(request->con);
		return;
	}

	if(err != 0) {
		error_log("fd %d connect failed\n", fd);
		return;
	}

	debug_log(3, "fd %d connected\n", fd);
	http_request_send(request);
}

static int conn_server_fd_create(request_t *req)
{
	assert(req);

	conn_t *con = req->con;
	int h_error;
	char buf[1024];
	struct hostent *srv_host, host_info;
	int server_fd = sock_open(1, 1);
	char server_ip[16];

	if(server_fd == -1) {
		error_log("fd %d, failed to open new socket\n", server_fd);
		conn_free(con);
		return -1;
	}

	if(0 != gethostbyname_r((char *)req->host, &host_info, buf, sizeof(buf), &srv_host, &h_error )) {
		debug_log(1, "%s host search failed. {%s}\n", req->host, strerror(h_error));
		sock_close(server_fd);
		conn_free(con);
		return -1;
	}
	if(srv_host == NULL) {
		debug_log(1, "host %s gethostbyname failed. {%s}\n", \
					   	req->host, strerror(h_error));
		sock_close(server_fd);
		conn_free(con);
		return -1;
	}
	debug_log(5, "host %s(%s)\n", req->host, server_ip);
	if(NULL == (char *)inet_ntop(AF_INET, srv_host->h_addr, server_ip, 16)) {
		error_log("fd %d inet_ntop failed\n", server_fd);
		sock_close(server_fd);
		conn_free(con);
		return -1;
	}
		
	if(inet_pton(AF_INET, server_ip, &con->server_addr) < 0) {
		error_log("fd %d inet_pton failed\n", server_fd);
		sock_close(server_fd);
		conn_free(con);
		return -1;
	}
	con->sfd = server_fd;
	
	debug_log(3, "success create %d\n", server_fd);
	return server_fd;
}

static void http_request_send(request_t *request)
{
	if(!request) 
		return;
	
	string *out;
	conn_t *con = request->con;
	
	out = request_string_build(request);
	if(out == NULL) {
		if(cache_lock())
			return;
		request->cache->statu = CACHE_DOWNLOAD_FAILED;
		cache_file_delete_and_free(request->cache);
		cache_unlock();
		conn_free(con);
		return;
	}
	/* append body later */
	sock_write(&con->server_event, con->sfd, http_request_send_finish, request, out);	
	string_clean(out);
}


string *request_string_build(const request_t *request) 
{
	if(!request)
		return NULL;
	
	string *retv = string_limit_init(MAX_HEADER);
	int res;
	char line[MAX_BUFFER];
	char ip[16];
	int i;
	header_entry_t *e;
	header_t *h = request->headers;
	struct in_addr *client;

	res = snprintf(retv->buf, MAX_HEADER, HttpRequestLineFormat, \
				request->method.name, request->url_path, request->version.major, request->version.minor);
	retv->offset += res;
	
	for(i = 0; i < h->entries.count; i++) {
		e = h->entries.items[i];
		assert(e);
		/* ignore Connection */
		if(e->id == HDR_CONNECTION) 
			continue;
		if(e->id == HDR_RANGE)
			continue;
		if(e->id == HDR_IF_MODIFIED_SINCE)
			continue;
		if(e->id == HDR_IF_NONE_MATCH)
			continue;
		if(e->id == HDR_PROXY_CONNECTION)
			continue;
		string_append(retv, e->name, strlen(e->name));
		string_append(retv, ": ", 2);
		string_append(retv, e->value, strlen(e->value));
		string_append(retv, "\r\n", 2);
	}
	/* handle revalidate */
	if(request->hit_state == TCP_REVALIDATE) {
		/* support ETag later */
		cache_file_t *cache = request->cache;
		assert(cache);
		if(cache->time.last_modified > 0) {
			res = snprintf(line, MAX_BUFFER - 1, "If-Modified-Since: %s\r\n", mkrfc1123(cache->time.last_modified));
		}
		else {
			debug_log(3, "last modfied %ld <= 0, use Date time.\n", cache->time.last_modified);
			res = snprintf(line, MAX_BUFFER - 1, "If-Modified-Since: %s\r\n", mkrfc1123(cache->time.date));
		}
		string_append(retv, line, res);
	}
	/* append Via */
	res = snprintf(line, MAX_BUFFER -1, "Via: %s\r\n", PACKAGE_STRING);
	string_append(retv, line, res);
	/* append x-forward-for */
	if(request->con) 
		client = &request->con->client_addr;
	if(client && inet_ntop(AF_INET, client, ip, 16)) {
		res = snprintf(line, MAX_BUFFER - 1, "X-Forward-For: %s\r\n", ip);
		string_append(retv, line, res);
	}
	/* append Connection */
	res = snprintf(line, MAX_BUFFER -1, "Connection: Keep-Alive\r\n");
	string_append(retv, line, res);
	string_append(retv, "\r\n", 2);	
	/* append body */
	if(request->body) {
		string_append_r(retv, request->body);
	}

	debug_log(9, "\n======\n%s\n======\n}\n", retv->buf);
	return retv;
}

void conn_timeout_handle(int fd, short flag, void *arg) 
{
	conn_t *con = arg;

	if(!con) 
		return;

	debug_log(2, "fd %d timeout\n", con->cfd);
	conn_free(con);
}

static void http_request_send_finish(int fd, int error, int res, void *arg)
{
	assert(arg);
	
	request_t *request = arg;
	conn_t *con = request->con;
	cache_file_t *cache = request->cache;
	struct timeval tv;

	if(error) {
		error_log("fd %d write failed\n", fd);
		if(cache_lock()) 
			return;
		cache->statu = CACHE_DOWNLOAD_FAILED;
		cache_file_delete_and_free(cache);
		cache_unlock();
		conn_free(con);
		return;
	}
	
	debug_log(3, "send %d bytes\n", res);
	if(-1 == gettimeofday(&tv, NULL)) {
		error_log("fd %d failed to get time\n", fd);
	}
	/* record request time, not need to lock */
	cache->time.request_issued = tv.tv_sec;
	/* update timeout */
	timeout_update(&con->timeout_event, conn_timeout_handle, con, conf->timeout.reply);
	if(request->hit_state == TCP_MISS) {
		/* update server event to read reply */
		if(-1 == event_update(&con->server_event, con->sfd, EV_READ | EV_PERSIST, http_reply_read, request))
			error_log("update %d server event failed\n", con->sfd);
	}
	else if(request->hit_state == TCP_REVALIDATE) {
		event_update(&con->server_event, con->sfd, EV_READ | EV_PERSIST, http_ims_reply_read, request);
	}
}

static void http_reply_read(int fd, short flag, void *arg)
{
	/* read data from server until completely */
	assert(arg);
#define SOCK_BUF 4096
	char buf[SOCK_BUF];
	request_t *r = arg;
	conn_t *con = r->con;
	reply_t *rep;
	int left;
	int sz = 0;
	cache_file_t *cache = r->cache;
	int cacheable = CACHE_ABLE_YES;

	rep = (r->reply == NULL) ? (r->reply = reply_init()) : r->reply;
	sz = read(fd, buf, SOCK_BUF);
	if(sz == -1) {
		if(error_ignore(errno)) 
			return;
		
		error_log("fd %d reply read failed\n", fd);
		sock_close(fd);
		event_del(&con->server_event);
		return;
	}
	else if(sz == 0 && rep->body_in == 0) {
		/* free con and delete cache from hashtable */
		cache_lock();
		cache->statu = CACHE_DOWNLOAD_FAILED;
		cache_file_delete_and_free(cache);
		cache_unlock();
		conn_free(con);
		return;
	}
	else if(sz == 0 && rep->body_size > 0) {
		debug_log(6, "%d data in. body in %d, body size %d\n", sz, rep->body_in, rep->body_size);
		if(rep->body_in < rep->body_size) {
			return;
		}
	}
	else if(sz == 0) {
		/* we consider that object has downloaded completely. 
		 * it's time out check reply->in. Do need to swap out to disk ?
		 */
		if(rep->swapout.offset < rep->body_in + rep->header_size) {
			if( -1 == http_reply_swapout(rep, cache->file)) {
				cache_lock();
				cache->statu = CACHE_DOWNLOAD_FAILED;
				cache_file_delete_and_free(cache);
				cache_unlock();
				conn_free(con);
				return;
			}
		}
		if(rep->body_size == 0)
			rep->body_size = rep->body_in;
		cache_file_download_success(cache, rep->header_size, rep->body_size);
		event_del(&con->server_event);
		sock_close(con->sfd);
		debug_log(3, "%s download finish. body_size %d, body_in %d\n", cache->file->buf, rep->body_size, rep->body_in);
		return;
	}
	mem_append(rep->in, buf, sz);
	http_reply_swapout(rep, cache->file);
	if(!rep->parsed) {
		timeout_update(&con->timeout_event, conn_timeout_handle, con, conf->timeout.server_life);
		if(-1 == reply_parse(rep, buf, sz)) {
			debug_log(1, "parse reply %p failed\n", rep);
			return;
		}
		if(cache_lock())
			return;
		cache_control_set(&rep->cc, rep->headers);
		cache_time_set(r, cache);
		cache_unlock();
		cacheable = r->cacheable;
		if(cacheable == CACHE_ABLE_YES) 
			cacheable = cache_able(r);
		if(cacheable == CACHE_ABLE_NO) {
			if(cache_lock())
				return;
			cache_file_delete_and_free(cache);
			cache_unlock();
		}
		if( -1 == event_update(&con->client_event, con->cfd, EV_WRITE | EV_PERSIST, http_reply_send, r)) {
			error_log("fd %d event update failed\n", con->cfd);
			return;
		}
	}
	else 
		rep->body_in += sz;

	debug_log(3, "%s download %d, total{header %d, body %d}\n", \
					cache->file->buf, sz, rep->header_size, rep->body_in);
	if(rep->body_size > 0 && rep->body_in >= rep->body_size) {
		if ( -1 == http_reply_swapout(rep, cache->file)) {
			/* swap out failed */
			if(cache_lock()) 
				return;
			cache->statu == CACHE_DOWNLOAD_FAILED;
			cache_file_delete_and_free(cache);
			cache_unlock();
			conn_free(con);
			return;
		}
		
		event_del(&con->server_event);
		sock_close(con->sfd);
		cache_file_download_success(cache, rep->header_size, rep->body_size);
		return;
	}
}

static reply_t * reply_init(void)
{
	reply_t *reply = mempool_alloc(sizeof(*reply));
	assert(reply);

	memset(reply, 0, sizeof(*reply));
	reply->in = mem_init(conf->server.max_object_size_in_mem);
	reply->swapout.offset = 0;
	reply->disk = -1;

	return reply;
}

static int reply_parse(reply_t *reply, const char *buf, const size_t size)
{
	assert(reply);
	assert(buf);
	assert(size > 0);
	
	char sline[MAX_BUFFER];
	char *start, *end;
	char *token;
	int s_len;
	int header_len;
	char *tmp;
	
	if(reply->parsed == 1) 
		return 0;
	
	start = (char *)buf;
	if(NULL == (end = strchr(start, '\n'))) {
		debug_log(1, "%p uncompleted header\n", reply);
		return -1;
	}

	if(0 != strncasecmp(start, "HTTP/", 5)) {
		debug_log(1, "%p unsupport reply\n", reply);
		return -1;
	}

	s_len = end - start;
	assert(s_len > 0);
	
	snprintf(sline, s_len + 1, "%s",  start);  /* include '\n' */
	/* get http version */
	tmp = sline;
	token = (char *)strsep(&tmp, " \t\r\n");
	if(token == NULL) {
		debug_log(1, "%p uncompleted header\n", reply);
		return -1;
	}
	reply->version.major = atoi(token + 5);
	if(NULL == (start = strchr(token, '.'))) {
		debug_log(1, "%p unknown http version\n", reply);
		return -1;
	}
	reply->version.minor = atoi(start + 1);
	/* get http status */
	if(NULL == (token = (char *)strsep(&tmp, " \t\r\n"))) {
		debug_log(1, "%p not find statu code\n", reply);
		return -1;
	}
	reply->statu = atoi(token);
	debug_log(3, "statu %d\n", reply->statu);

	/* begin to parse header */
	start = end + 1;
	header_len = headers_end(start, size - s_len - 1);
	reply->headers = http_header_parse(start, header_len + 1);
	reply->header_size = headers_end(buf, size); /* include statu line */
	reply->body_in = size - reply->header_size;
	if(reply->headers) {
		reply->parsed = 1;
		header_entry_t *e = header_entry_get(reply->headers, HDR_CONTENT_LENGTH);
		if(e) {
			reply->body_size = atoi(e->value);
			debug_log(3, "header %d, body %d\n", \
							reply->header_size, reply->body_size);
		}
		else {
			debug_log(3, "%p not contain Content-Length\n", reply);
			reply->body_size = 0;
		}
		return 0;
	}
	else 
		return -1;
}

string *reply_header_build(header_t *head, int statu, http_version_t version, int conn_state)
{
	assert(head);

	int i;
	header_entry_t *e;
	header_t *h = head;
	int header_sz = 0;
	string *retv = string_limit_init(MAX_BUFFER);
	char sline[MAX_BUFFER];
	int res;

	res = snprintf(sline, MAX_BUFFER - 1, HttpReplyLineFormat, \
			version.major, version.minor, statu, http_status_string(statu));
	
	string_append(retv, sline, res);
	for(i = 0; i < h->entries.count; i++) {
		e = h->entries.items[i];
		assert(e);
		if(e->id == HDR_CONNECTION) 
			continue;
		string_append(retv, e->name, strlen(e->name));
		string_append(retv, ": ", 2);
		string_append(retv, e->value, strlen(e->value));
		string_append(retv, "\r\n", 2);
	}
	/* add mark */
	res = snprintf(sline, MAX_BUFFER - 1, "Via: %s %s\r\n", hit_state_string(conn_state), PACKAGE_STRING);
	string_append(retv, sline, res);
	/* append Connection */
	res = snprintf(sline, MAX_BUFFER - 1, "Connection: close\r\n");
	string_append(retv, sline, res);

	string_append(retv, "\r\n", 2);	
	return retv;
}

static void http_reply_send(int fd, short flag, void *data)
{
	if(!data)
		return;
	
	cache_file_t * cache;
	reply_t *rep ;
	mem_t *mem_cache;
	shm_t *shm_cache;
	request_t *r = data;
	conn_t *con = r->con;
	
	assert(r);
	cache = r->cache;
	assert(cache);
	rep = r->reply;
	mem_cache = cache->mem_cache;
	shm_cache = cache->shm_cache;
	
	if(	r->hit_state == TCP_MISS) {
		if(!rep) 
			return;
		if(!rep->header_send) {
			http_reply_header_send(fd, r, DATA_REPLY_IN);
			return;
		}
		/* check body size */
		int data_state = required_data_is_exist(rep);
		if(data_state == DATA_NONE) {
			debug_log(2, "data[%d-%d] not find\n", rep->body_out + rep->header_size, rep->body_in + rep->header_size);
			return;
		}
		
		debug_log(3, "%s, %s, %s\n", \
						cache->file->buf, hit_state_string(r->hit_state), data_state_to_string(data_state));	
		http_reply_body_send(fd, r, data_state);
		return;
	}
	
	/* HIT state. */
	int data_state = DATA_NONE;
	int statu; 

	if(cache_lock())
		return;
	statu = cache->statu;
	cache_unlock();
	if(statu == CACHE_DOWNLOADING) {
		debug_log(3, "cache %s not downloaded now\n", cache->file->buf);
		return;
	}
	else {
		if(rep->header_size == 0 || rep->body_size == 0) {
			if(cache_lock())
				return;
			rep->header_size = cache->header_size;
			rep->body_size = cache->body_size;
			cache_unlock();
		}
	}
	
	debug_log(3, "TCP_HIT reply header %d, body %d\n", rep->header_size, rep->body_size);	
	if(rep->header_size <= 0 || rep->body_size <= 0)
		return;
	if(!rep->header_send) {
		http_reply_header_send(fd, r, DATA_REPLY_IN);
		return;
	}

	data_state = required_data_is_exist(rep);	
	if(data_state == DATA_NONE) 
		return;
		
	debug_log(3, "%s data state %s\n", hit_state_string(r->hit_state), data_state_to_string(data_state));	
	http_reply_body_send(fd, r, data_state);
	return;

}

static int http_reply_swapout(reply_t *rep, string *file)
{
	assert(file);
	if(!rep) 
		return 0;

	int res;
	int offset = rep->swapout.offset;
	int fd = rep->disk;
	int size;
	int buf_offset;

	if(fd < 0) {
		fd = open(file->buf, O_WRONLY | O_CREAT, 0666);
		if(fd < 0) {
			error_log("open %s failed\n", file->buf);
			return -1;
		}
		rep->disk = fd;
	}
	if(rep->swapout.offset >= rep->in->high) 
		return 0;
	else if(rep->swapout.offset < rep->in->low) {
		warn_log("swapout %d, low %d\n", rep->swapout.offset, rep->in->high);
		return 0;
	}
	size = rep->in->high - rep->swapout.offset;
	buf_offset = rep->in->inmem->offset - size;
	res = write(fd, rep->in->inmem->buf + buf_offset, size);
	if(res == -1) {
		error_log("write %d failed\n", fd);
		return;
	}
	rep->swapout.offset += res;
	/* maintain mem */
	reply_in_mem_maintain(rep);
	debug_log(6, "result %d, swapout buf %p, buf offset %d, size %d, file offset %d ==> %d\n", 
					res, rep->in->inmem->buf + buf_offset, buf_offset, size, offset, rep->swapout.offset);
	return res;
}

static void http_reply_header_send(int fd, request_t *r, int data_state)
{
	assert(r);
	assert(r->reply);
	
	int state = r->hit_state;
	conn_t *con = r->con;
	cache_file_t *cache = r->cache;
	reply_t *rep = r->reply;
	string *hstr;
	
	if(state == TCP_MISS) {
		if(!rep) {
			debug_log(3, "state %s and rep %p\n, return\n", hit_state_string(state), rep);
			return;
		}
		else {
			if(!rep->headers) {
				debug_log(3, "rep->headers %p , return\n", rep->headers);
				return;
			}
			else 
				hstr = reply_header_build(rep->headers, rep->statu, rep->version, state);
		}
	}
	else { 
		/* HIT state */
		if(rep->headers) 
			hstr = reply_header_build(rep->headers, rep->statu, rep->version, state);
		else {
			if(rep->disk == -1) {
				rep->disk = open(cache->file->buf, O_RDONLY);
				if(rep->disk == -1) 
					return;
			}
			char buf[MAX_BUFFER];
			int sz;
			assert(rep->header_size > 0);
			assert(rep->header_size < MAX_BUFFER);
			sz = pread(rep->disk, buf, rep->header_size, 0);
			if(-1 == sz) 
				return;
			if( -1 == reply_parse(rep, buf, sz)) {
				return;
			}
			hstr = reply_header_build(rep->headers, rep->statu, rep->version, state);
		}
	}

	assert(hstr);
	sock_write(&con->client_event, con->cfd, http_reply_header_send_done, r, hstr);
	string_clean(hstr);
}

static void http_reply_header_send_done(int fd, int error, int res, void *arg)
{
	assert(arg);
	request_t *r = arg;
	reply_t *rep = r->reply;
	conn_t *con = r->con;
	
	if(error) {
		error_log("fd %d header send failed\n", fd);
		conn_free(con);
		return;
	}

	assert(res >= 0);
	rep->header_out += res;
	rep->header_send = 1;
	debug_log(3, "%d bytes header out\n", res);
	if( -1 == event_update(&con->client_event, con->cfd, EV_WRITE | EV_PERSIST, http_reply_send, r))
		error_log("fd %d event update failed\n", con->cfd);
	http_reply_send(fd, EV_WRITE, r);
}

static void http_reply_body_send(int fd, request_t *r, int data_state)
{
	assert(r);
	
	int size;
	char *buf;
	int buf_offset;
	int low, high;
	conn_t *con = r->con;
	reply_t *rep = r->reply;
	cache_file_t *cache = r->cache;
	mem_t *mc ;
	shm_t *sc ;
	mmap_t *map ;
	mem_t *in;
	int pos;
	
	assert(cache);
	assert(rep);
	in = rep->in;
	if(data_state == DATA_REPLY_IN) {
		string *str ;
		
		low = in->low;
		high = in->high;
		pos = rep->body_out + rep->header_size;
		debug_log(8, "reply body size %d, [%d-%d], pos %d\n", \
					   	rep->body_size, low, high, pos);
		if(high <= pos)
			return;
		if(low > pos)
			return;
		size = high - pos;
		buf_offset = in->inmem->offset - size;
		buf = in->inmem->buf + buf_offset;
		str = string_init(buf, size);
		sock_write(&con->client_event, con->cfd, http_reply_body_send_done, r, str);
		string_clean(str);
		return;
	}
	else if(data_state == DATA_MEM_CACHE) {
	}
	else if(data_state == DATA_SHM_CACHE) {
		return;
	}
	else if(data_state == DATA_DISK) {
		pos = rep->body_out + rep->header_size;
		high = rep->swapout.offset;
		if(pos >= high) 
			return;
		debug_log(8, "reply body size %d, disk %d, pos %d\n", \
					   	rep->body_size, high, pos);
		size = high - pos;
		assert(size >= 0);
		size = XMIN(DEFAULT_MAP_SIZE, size);
		if(-1 == disk_map(rep, cache->file, pos, high)) {
			debug_log(3, "map [%d - %d] failed\n", pos, high);
			return;
		}
		map = &rep->map;
		sock_sendfile(&con->client_event,  map->shm_fd, con->cfd, 0, size, http_reply_body_send_done, r);
		return;
	}
}

static void http_reply_body_send_done(int fd, int error, int res, void *arg)
{
	assert(arg);
	
	request_t *r = arg;
	conn_t *con = r->con;
	cache_file_t *cache = r->cache;
	reply_t *rep = r->reply;

	assert(cache);
	assert(rep);
	assert(res >= 0);
	if(error) {
		error_log("fd %d reply body send failed\n", fd);
		if(cache_lock()) 
			return;
		cache->statu = CACHE_DOWNLOAD_FAILED;
		cache_file_delete_and_free(cache);
		cache_unlock();
		conn_free(con);
		return;
	}
	rep->body_out += res;
	debug_log(3, "%s send %d, total{H %d, B %d}\n", cache->file->buf, res, rep->header_out, rep->body_out);
	if(rep->body_size <= 0) {
		if(-1 == event_update(&con->client_event, con->cfd, EV_WRITE | EV_PERSIST, http_reply_send, r))
			error_log("fd %d event update failed\n", con->cfd);
		return;
	}
	if(rep->body_out >= rep->body_size) {
		/* send finish */
		debug_log(3, "%s send finish %d bytes.\n", cache->file->buf, rep->body_out);
		conn_free(con);
		return;
	}
	if( -1 == event_update(&con->client_event, con->cfd, EV_WRITE | EV_PERSIST, http_reply_send, r))
		error_log("fd %d event update failed\n", con->cfd);
}

int mem_cache_find(request_t *r, cache_file_t *cache)
{
	return 0;
}

int shm_cache_find(request_t *r, cache_file_t *cache)
{
	return 0;
}

int disk_find(request_t *r, cache_file_t *cache)
{
	assert(r);
	assert(cache);

	if(cache->statu == CACHE_IN_DISK) 
		return 1;
	else  if(r->hit_state == TCP_MISS) {
		if(r->reply->body_out < r->reply->swapout.offset)
			return 1;
		else 
			return 0;
	}
	else 
		return 0;
}

char *data_state_to_string(int state)
{
	switch(state) {
		case DATA_NONE :
			return "DATA_NONE";
		case DATA_REPLY_IN:
			return "DATA_REPLY_IN";
		case DATA_MEM_CACHE:
			return "DATA_MEM_CACHE";
		case DATA_SHM_CACHE:
			return "DATA_SHM_CACHE";
		case DATA_DISK:
			return "DATA_DISK";
		default :
			return "DATA_UNKOWN";
	}
}

int disk_map(reply_t *rep, string *file, int low, int high)
{
	assert(rep);

	mmap_t *map = &rep->map;
	char temp_file[32];
	int fd;
	int need;
	
	assert(map);
	if(high - low <= 0)  {
		debug_log(4, "[%d - %d] <= 0, return.\n", low, high);
		return -1;
	}
	if(!file) {
		debug_log(4, "%s disk map failed\n", file->buf);
		return -1;
	}
	
	if(rep->disk < 0) {
		rep->disk = open(file->buf, O_RDONLY);
		if(rep->disk < 0) {
			error_log("%s open failed\n", file->buf);
			return -1;
		}
	}

	disk_map_reset(&rep->map);

	strcpy(temp_file, "/dev/shm/c-XXXXXX");
	if(-1 == (fd = mkstemp(temp_file))) {
		error_log("temp_file %s open failed\n", temp_file);
		return -1;
	}
	if(-1 == unlink(temp_file))
		warn_log("%s unlink failed\n", temp_file);

	need = high - low;

	ftruncate(fd, need);

	map->start = mmap(NULL, need, PROT_WRITE | PROT_READ, MAP_SHARED, fd, 0);
	if(MAP_FAILED == map->start) {
			error_log("fd %d, bytes %d, offset %lld map failed\n", \
							fd, need, 0);
			close(fd);
			return -1;
	}
	if( -1 == madvise(map->start, need, MADV_NORMAL)) {
			error_log("%s madvise failed\n", temp_file);
	}

	map->shm_fd = fd;
	map->size = need;
	map->low = low;
	map->high = high;
	
	debug_log(3, "success map %s [%d-%d] in %s\n", file->buf, low, high, temp_file);
	return need;
}

void disk_map_reset(mmap_t *map)
{
	if(!map) 
		return;
	
	if(map->start)
		munmap(map->start, map->size);
	if(map->shm_fd) 
		close(map->shm_fd);

	map->start = NULL;
	map->low = 0;
	map->high = 0;
	map->size = 0;
	map->shm_fd = -1;
}

void disk_map_free(mmap_t *map)
{
	if(!map) 
		return;
	
	if(map->start)
		munmap(map->start, map->size);
	if(map->shm_fd) 
		close(map->shm_fd);
	
	mempool_free(map);
}

static void http_ims_reply_read(int fd, short flag, void *arg)
{
	/* read data from server until completely */
	assert(arg);
#define SOCK_BUF 4096
	char buf[SOCK_BUF];
	request_t *r = arg;
	conn_t *con = r->con;
	reply_t *rep;
	int left;
	int sz = 0;
	string *in ;
	cache_file_t *cache = r->cache;

	
	sz = read(fd, buf, SOCK_BUF);
	
	if(sz == -1) {
		if(error_ignore(errno)) 
			return;
		
		error_log("fd %d reply read failed\n", fd);
		conn_free(con);
		return;
	}
	else if(sz == 0 && rep->body_in == 0) {
		conn_free(con);
		return;
	}
	else if(sz == 0) {
		/* we consider that object has downloaded completely. 
		 * it's time out check reply->in. Do need to swap out to disk ?
		 */
		if(rep->swapout.offset < rep->body_in + rep->header_size) {
			if( -1 == http_reply_swapout(rep, cache->file)) {
				cache_lock();
				cache->statu = CACHE_DOWNLOAD_FAILED;
				cache_file_delete_and_free(cache);
				cache_unlock();
				conn_free(con);
				return;
			}
		}
		if(rep->body_size == 0)
			rep->body_size = rep->body_in;
		
		debug_log(3, "reply->body_size %d, reply->body_in %d\n", rep->body_size, rep->body_in);
		
		if(cache_lock()) 
			return;
		cache->statu = CACHE_IN_DISK;
		cache_unlock();
		return;
	}
	
	if(rep->body_in == 0 && sz > 0) 
		timeout_update(&con->timeout_event, conn_timeout_handle, con, conf->timeout.server_life);
	
	if(!rep->parsed) {
		if(-1 == reply_parse(rep, buf, sz)) {
			return;
		}
		
		if(rep->statu == HTTP_NOT_MODIFIED) {
			if(cache_lock()) 
				return;
			cache_time_set(r, cache);
			cache_unlock();
			
			event_del(&con->server_event);
			sock_close(con->sfd);
			reply_free(rep);
			r->reply = NULL;
			http_hit_handle(r, TCP_HIT);
			return;
		}
		else if(rep->statu == HTTP_OK) {
			if(cache_lock()) 
				return;
			cache_file_reset(cache);
			cache_time_set(r, cache);
			cache_unlock();

			cache_control_set(&rep->cc, rep->headers);
			/* update client event */
			if ( -1 == event_update(&con->client_event, con->cfd, EV_WRITE | EV_PERSIST, http_reply_send, r)) 
				debug_log(3, "fd %d update event failed\n", con->cfd);
			
			debug_log(3, "fd %d update event\n", con->cfd);
			r->hit_state = TCP_MISS;
		}
		else {
			debug_log(4, "%d statu reply from server invalid\n", rep->statu);
			conn_free(con);
			return;
		}
	}
	else 
		rep->body_in += sz;

	debug_log(3, "%s download %d, total{header %d, body %d}\n", \
					cache->file->buf, sz, rep->header_size, rep->body_in);

	if(rep->body_size > 0 && rep->body_in >= rep->body_size) {
		debug_log(3, "%s download finish!\n", cache->file->buf);
		if ( -1 == http_reply_swapout(rep, cache->file)) {
			/* swap out failed */
			if(cache_lock()) 
				return;
			cache->statu == CACHE_DOWNLOAD_FAILED;
			cache_file_delete_and_free(cache);
			cache_unlock();
			conn_free(con);
			return;
		}
		
		event_del(&con->server_event);
		sock_close(con->sfd);
		if(cache_lock()) 
			return;
		cache->statu = CACHE_IN_DISK;
		cache_unlock();
		return;
	}
}


static void http_hit_waiting(int fd, short flag, void *data)
{
	assert(data);

	request_t *r = data;
	conn_t *con = r->con;
	reply_t * rep = r->reply;
	cache_file_t *cache = r->cache;
	int statu;
	
	assert(rep);
	if(cache_lock()) 
		return;	
	rep->header_size = cache->header_size;
	rep->body_size = cache->body_size;
	statu = cache->statu;
	cache_unlock();
	
	debug_log(9, "statu %s, header_size %d, body_size %d\n", cache_statu_to_string(statu), \
					rep->header_size, rep->body_size);
	if(statu == CACHE_DOWNLOAD_FAILED) {
		conn_free(con);
		return;
	}
	if(rep->header_size == 0 && rep->body_size == 0) {
		timer_update(&con->timer, http_hit_waiting, r, 0, 300);
		debug_log(3, "timer %d.%d callback\n", 0, 300);
		return;
	}	
	else if(event_update(&con->client_event, con->cfd, EV_WRITE | EV_PERSIST, http_reply_send, r) == -1) {
		debug_log(3, "fd %d event update failed\n", con->cfd);
		return ;
	}

	/* del timer */
	evtimer_del(&con->timer);
	
}

static void http_request_read(int fd, short flag, void *arg)
{
	assert(arg);

	request_t *r = arg;
	conn_t *con = r->con;
	string *in = con->in;
	string *body = r->body;
	int left;
	int res;

	/* body read */
	assert(body);
	left = in->size - in->offset;
	if(left <= 0)  {
		string_reset(in);
		left = in->size - in->offset;
	}
	res = read(fd, in->buf + in->offset, left);
	if(res < 0) {
		if(!error_ignore(errno))
			conn_free(con);
		return;
	}
	else if(res == 0 && body->offset == 0) {
		debug_log(5, "fd %d no body in\n", body->offset);
		conn_free(con);
		return;
	}
	string_append(body, in->buf + in->offset, res);
	in->offset += res;
	if(body->offset < r->content_length) {
		debug_log(5, "request body offset %d, length %d\n", body->offset, r->content_length);
		return;
	}

	debug_log(5, "request body %d received all.\n", r->content_length);
	http_request_handle(r);
}

static int required_data_is_exist(reply_t *reply)
{
	assert(reply);
	
	int low, high;
	int pos;
	int state = DATA_NONE;

	if(reply->header_send) 
		pos = reply->header_size + reply->body_out;
	else 
		pos = 0;
	/* is in mem */
	low = reply->in->low;
	high = reply->in->high;
	assert(high - low == reply->in->inmem->offset);
	if(pos < high && pos >= low) { 
		state = DATA_REPLY_IN;
		goto result;
	}
	/* is in disk ? */
	low = 0;
	high = reply->swapout.offset;
	if(pos < high && pos >= low) {
		state = DATA_DISK;
		goto result;
	}

result:
	debug_log(6, "data in %s\n", data_state_to_string(state));
	return state;
}

static void reply_in_mem_maintain(reply_t *rep) 
{
	if(!rep)
		return;

	int low, high;
	int disk_pos;
	
	low = rep->in->low;
	high = rep->in->high;
	disk_pos = rep->swapout.offset;
	if(disk_pos < low) 
		assert(0);
	else if(disk_pos == low)
		return;
	assert(disk_pos > low);
	if(disk_pos >= high) {
		/* free [low, high] */                
		string_reset(rep->in->inmem); 
		rep->in->low = rep->in->high;                
		debug_log(6, "inmem %d, mem [%d, %d]\n", rep->in->inmem->offset, rep->in->low, rep->in->high);
		return;
	}
	assert(disk_pos < high) ;
	/* free [low, disk_pos] */
	int size = high - disk_pos;
	int offset = rep->in->inmem->offset - size;
	string *new = string_limit_init(size);
	string_append(new, rep->in->inmem->buf + offset, size);
	string_clean(rep->in->inmem);
	rep->in->inmem = new;
	rep->in->low = disk_pos;
	debug_log(6, "inmem %d, mem [%d, %d]\n", rep->in->inmem->offset, rep->in->low, rep->in->high);
}
